package com.gitee.source

import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
/*
  并行数据源:
    在3个并行里面同时每个一秒打印 n+1
  1. 实现ParallelSourceFunction接口
 */
object CustomerSourceWithParallel {
  def main(args: Array[String]): Unit = {
    //1.创建senv
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //2.从自定义的数据源读取数据
    import  org.apache.flink.api.scala._
    val data: DataStream[Long] = senv.addSource(new MyParallel()).setParallelism(3)
    data.print()
    senv.execute()
  }

}

class MyParallel extends ParallelSourceFunction[Long]() {
  var currentNum:Long = 1
  var isRunning:Boolean = true//表示是否正在执行

  override def cancel(): Unit = {
    isRunning = false
  }
  //run方法就是用来生成数据的
  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
    while(isRunning){
      //使用ctx将数据生成出去
      ctx.collect(currentNum)
      currentNum+=1
      Thread.sleep(1000)
    }
  }
}
